package com.pactera.asmp.server.common.websocket;

import com.pactera.asmp.server.pojo.SocketMessage;
import com.pactera.asmp.server.utils.DateUtil;

import lombok.Synchronized;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

import javax.websocket.Session;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
//@RabbitListener(queues = "#{directQueue.name}")
public class DirectReceiver {
    @Autowired
    private StringRedisTemplate redisTemplate;

    private static ConcurrentHashMap<String, ReentrantLock> lockMap = new ConcurrentHashMap<>();

//    @RabbitHandler
    public void process(SocketMessage message) {
        try {
            if (message == null) return;
//            if("json".equalsIgnoreCase(message.getType())){
//                log.info("{} --- DirectReceiver消费者收到消息: from-{},to-{},type-{},data-{}", DateUtil.now(), message.getFrom(), message.getTo(), message.getType(),new String(message.getData(),"utf-8"));
//            }else {
//                log.info("{} --- DirectReceiver消费者收到消息: from-{},to-{},type-{}", DateUtil.now(), message.getFrom(), message.getTo(), message.getType());
//            }
//            if("timeout".equalsIgnoreCase(message.getType())){
//                log.info("{} --- DirectReceiver消费者收到消息: from-{},to-{},type-{},data-{}", DateUtil.now(), message.getFrom(), message.getTo(), message.getType(),new String(message.getData(),"utf-8"));
//            }
            String toId = message.getTo();
            if (!StringUtils.isEmpty(toId)) {
                // 根据to找到相应的socket
                String instanceKey ;
                ConcurrentHashMap<String, Session> id2sessionMap;
                String duid = message.getFrom().split("@")[0];
                if(toId.startsWith("screen:")){
                    instanceKey =  SocketUtil.REDIS_KEY_WS_SCREEN_INSTANCE + ":" + toId;
                    id2sessionMap = WebSocketServerScreen.id2sessionMap;
                }
                else if(toId.length() == 36){
                    instanceKey =  SocketUtil.REDIS_KEY_WS_DEVICE_INSTANCE + ":" + toId;
                    id2sessionMap = WebSocketServerDevice.id2sessionMap;
                }
                else {
                    instanceKey =  SocketUtil.REDIS_KEY_WS_USER_INSTANCE + ":" + toId;
                    id2sessionMap = WebSocketServerUser.id2sessionMap;
                    if(toId.startsWith("CAN")){
                        id2sessionMap = WebSocketServerWebCan.id2sessionMap;
                    }

                    /*if(WebSocketServerUserAux.auxMap.get(duid)!=null){
                        id2sessionMap = WebSocketServerUserAux.id2sessionMap;
                        instanceKey = SocketUtil.REDIS_KEY_WS_USER_INSTANCE + ":" + toId+"aux";
                    }*/
                    if(toId.startsWith("aux")){
                        id2sessionMap = WebSocketServerUserAux.id2sessionMap;
                    }
                }

                Set<String> members = redisTemplate.opsForSet().members(instanceKey);

                if (members.isEmpty()) {
                    log.info("{} --- DirectReceiver找不到实例: {}", DateUtil.now(), instanceKey);
                    return;
                }
                if("timeout".equalsIgnoreCase(message.getType())){
                    id2sessionMap = WebSocketServerDevice.id2sessionMap;
                    if(toId.startsWith("CAN")){
                        id2sessionMap = WebSocketServerWebCan.id2sessionMap;
                    }
                    if(toId.startsWith("aux")){
                        id2sessionMap = WebSocketServerUserAux.id2sessionMap;
                    }
                    doTimeout(id2sessionMap,toId);  // 只有timeout时，toId是存在“@”
                    return;
                }
                ReentrantLock lock = lockMap.get(duid);
                if(lock == null){
                    synchronized(duid) {
                        lock = lockMap.get(duid);
                        if (lock == null) {
                            lockMap.put(duid, new ReentrantLock());
                            lock = lockMap.get(duid);
                        }
                    }
                }
               boolean isLock =  lock.tryLock();
                try {
                    if (isLock) {
                        this.handler(id2sessionMap, instanceKey, message, members);
                    }
                }finally {
                    if (isLock) {
                        lock.unlock();
                    }
                }
            }
        }catch (Exception e){
            log.info(e.getMessage());
        }
    }

    private void handler(ConcurrentHashMap<String, Session> id2sessionMap, String instanceKey, SocketMessage message, Set<String> members) throws IOException {
        Session session;
        for (String member : members) {
            session = id2sessionMap.get(member);
            if(session != null && session.isOpen()){
                if("img".equalsIgnoreCase(message.getType())){
                    session.getBasicRemote().sendBinary(ByteBuffer.wrap(message.getData()));
                }
                else {
                    session.getBasicRemote().sendText(new String(message.getData(), StandardCharsets.UTF_8));
                }
            }
            else {
                log.info("DirectReceiver找不到 {}-{}, REMOVE IT!!!",instanceKey,member);
                redisTemplate.opsForSet().remove(instanceKey, member);
            }
        }
    }

    private void doTimeout(ConcurrentHashMap<String, Session> id2sessionMap, String toId) {
        try{
            log.info("-------------doTimeout------------------{}",toId);
            Session session = id2sessionMap.get(toId);
            if (session != null && session.isOpen()) {
                log.info("DirectReceiver Close Timeout----------------{}",toId);
                session.close();
            }
            else {
                log.info("找不到相应的session-------{}",toId);
            }
        }
        catch (Exception e){
            log.info(e.getMessage());
        }
    }
}
 
